package io.kiki.sba.registry.server.clientInterface.push;


import io.kiki.sba.registry.common.model.DataCenterPushInfo;
import io.kiki.sba.registry.common.model.Tuple;
import io.kiki.sba.registry.common.model.constants.ValueConstants;
import io.kiki.sba.registry.common.model.store.DataCenterToSubDatumMap;
import io.kiki.sba.registry.common.model.store.SubDatum;
import io.kiki.sba.registry.concurrent.ThreadLocalStringBuilder;
import io.kiki.sba.registry.trace.TraceId;
import io.kiki.sba.registry.util.DatumVersionUtil;
import io.kiki.sba.registry.util.StringFormatter;
import org.apache.commons.collections.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

public final class Trace {
    // config the push trace log to use a separate disruptor
    private static final Logger logger = LoggerFactory.getLogger("PUSH-TRACE");

    final long pushCreateTimestamp = System.currentTimeMillis();
    final Cause cause;
    private final DataCenterToSubDatumMap dataCenterToSubDatumMap;
    private final String subApp;
    private final long subRegTimestamp;
    private final InetSocketAddress subAddress;
    private final int subNum;
    private volatile long pushStartTimestamp;

    private Trace(DataCenterToSubDatumMap dataCenterToSubDatumMap, InetSocketAddress address, String subApp, Cause cause, int subNum, long subRegTimestamp) {
        this.dataCenterToSubDatumMap = dataCenterToSubDatumMap;
        this.cause = cause;
        this.subAddress = address;
        this.subApp = subApp;
        this.subNum = subNum;
        this.subRegTimestamp = subRegTimestamp;
    }

    public static Trace trace(DataCenterToSubDatumMap datum, InetSocketAddress address, String subApp, Cause cause, int subNum, long subRegTimestamp) {
        return new Trace(datum, address, subApp, cause, subNum, subRegTimestamp);
    }

    static long getTriggerPushTimestamp(SubDatum subDatum) {
        long ts = subDatum.getVersion() <= ValueConstants.DEFAULT_NO_DATUM_VERSION ? System.currentTimeMillis() : DatumVersionUtil.getRealTimestamp(subDatum.getVersion());
        return ts;
    }

    private Tuple<List<Long>, String> datumPushedDelayList(String dataCenter, long finishedTs, long lastPushTs) {
        List<Long> recentVersions = dataCenterToSubDatumMap.getRecentVersions(dataCenter);
        List<Long> timestamps = new ArrayList<>(recentVersions == null ? 1 : recentVersions.size() + 1);
        StringBuilder builder = ThreadLocalStringBuilder.get();
        if (!CollectionUtils.isEmpty(recentVersions)) {
            for (long v : recentVersions) {
                long ts = DatumVersionUtil.getRealTimestamp(v);
                if (ts > lastPushTs) {
                    long delay = finishedTs - ts;
                    timestamps.add(delay);
                    builder.append(delay).append(',');
                }
            }
        }

        long datumChangeTs = DatumVersionUtil.getRealTimestamp(dataCenterToSubDatumMap.getDataCenterToSubDatumVersionMap(dataCenter));
        long delay = finishedTs - Math.max(lastPushTs, datumChangeTs);
        timestamps.add(delay);
        builder.append(delay);
        return new Tuple<>(timestamps, builder.toString());
    }

    public void startPush() {
        this.pushStartTimestamp = System.currentTimeMillis();
    }

    public void finishPush(PushStatus status, TraceId taskID, Map<String, DataCenterPushInfo> dataCenterPushInfoMap, int retry) {
        try {

            for (Entry<String, DataCenterPushInfo> entry : dataCenterPushInfoMap.entrySet()) {
                String dataCenter = entry.getKey();
                DataCenterPushInfo pushInfo = entry.getValue();
                finish(dataCenter, status, taskID, pushInfo, retry);
            }
        } catch (Throwable t) {
            logger.error("finish push error, {},{},{},{}", dataCenterToSubDatumMap.getDataInfoId(), dataCenterToSubDatumMap.getDataCenterToSubDatumVersionMap(), subAddress, taskID, t);
        }
    }

    private void finish(String dataCenter, PushStatus status, TraceId taskID, DataCenterPushInfo pushInfo, int retry) {
        final long subscriberPushedVersion = pushInfo.getPushVersion();

        final long pushFinishTimestamp = System.currentTimeMillis();
        // push.finish- first.newly.datumTimestamp(after subscriberPushedVersion)
        long datumModifyPushSpanMillis;
        // push.finish - datum.versionTs
        long datumVersionPushSpanMillis;
        // session.triggerTs - datum.versionTs
        long datumVersionTriggerSpanMillis;

        // task.create - session.triggerTs
        long pushTaskPrepareSpanMillis;
        // task.start - task.create
        long pushTaskQueueSpanMillis;
        // task.finish - task.start
        long pushTaskClientIOSpanMillis;
        // task.start - session.triggerTs
        long pushTaskSessionSpanMillis;

        // try find the earliest and the latest publisher after the subPushedVersion
        // that means the modify after last push, but this could not handle the publisher.remove
        final long lastTriggerSession = cause.triggerPushContext.getLastTraceTimes().getTriggerSession();
        if (cause.type == Type.reg) {
            datumVersionPushSpanMillis = pushFinishTimestamp - subRegTimestamp;
            datumVersionTriggerSpanMillis = Math.max(lastTriggerSession - subRegTimestamp, 0);
        } else {
            long datumTimestamp = cause.datumTimestamp.get(dataCenter);
            datumVersionPushSpanMillis = Math.max(pushFinishTimestamp - datumTimestamp, 0);
            datumVersionTriggerSpanMillis = Math.max(lastTriggerSession - datumTimestamp, 0);
            if (cause.type == Type.sub) {
                if (subRegTimestamp >= datumTimestamp) {
                    // case: datum.change trigger the sub.sub, but the sub.reg not finish
                    datumVersionPushSpanMillis = pushFinishTimestamp - subRegTimestamp;
                    datumVersionTriggerSpanMillis = Math.max(lastTriggerSession - subRegTimestamp, 0);
                }
            }
        }

        // calc the task span millis
        pushTaskPrepareSpanMillis = pushCreateTimestamp - cause.triggerPushContext.getFirstTraceTimes().getTriggerSession();
        pushTaskQueueSpanMillis = pushStartTimestamp - pushCreateTimestamp;
        pushTaskClientIOSpanMillis = pushFinishTimestamp - pushStartTimestamp;
        pushTaskSessionSpanMillis = pushStartTimestamp - cause.triggerPushContext.getFirstTraceTimes().getTriggerSession();

        final long lastPushTimestamp = subscriberPushedVersion <= ValueConstants.DEFAULT_NO_DATUM_VERSION ? subRegTimestamp : DatumVersionUtil.getRealTimestamp(subscriberPushedVersion);

        Tuple<List<Long>, String> datumPushedDelay = datumPushedDelayList(dataCenter, pushFinishTimestamp, lastPushTimestamp);
        List<Long> datumPushedDelayList = datumPushedDelay.o1;
        String pushDatumDelayStr = datumPushedDelay.o2;

        datumModifyPushSpanMillis = datumVersionPushSpanMillis;

        if (cause.type == Type.sub) {
            datumModifyPushSpanMillis = Math.max(datumPushedDelayList.get(0), datumVersionPushSpanMillis);
        }

        PushMetrics.Push.observePushDelayHistogram(dataCenter, cause.type, datumModifyPushSpanMillis, status);
        if (logger.isInfoEnabled()) {
            final String msg = StringFormatter.format("{},{},{},ver={},app={},cause={},pubNum={},pubBytes={},delay={},{},{},{},{}," + "session={},cliIO={}," + "subNum={},addr={},expectVer={},dataNode={},taskID={},pushedVer={},regTs={}," + "{},recentDelay={},pushNum={},retry={},encode={},encSize={}", status, dataCenterToSubDatumMap.getDataInfoId(), dataCenter, dataCenterToSubDatumMap.getDataCenterToSubDatumVersionMap(), subApp, cause.type, dataCenterToSubDatumMap.getPubNum(), dataCenterToSubDatumMap.getDataBoxBytes(), datumModifyPushSpanMillis, datumVersionPushSpanMillis, datumVersionTriggerSpanMillis, pushTaskPrepareSpanMillis, pushTaskQueueSpanMillis, pushTaskSessionSpanMillis, pushTaskClientIOSpanMillis, subNum, subAddress, cause.triggerPushContext.getExpectDatumVersion(), cause.triggerPushContext.dataNode, taskID, subscriberPushedVersion, subRegTimestamp, cause.triggerPushContext.formatTraceTimes(pushFinishTimestamp), pushDatumDelayStr, pushInfo.getPushNum(), retry, pushInfo.getEncode(), pushInfo.getEncodeSize());
            logger.info(msg);
            if (datumModifyPushSpanMillis > 6000) {
                logger.info(msg);
            }
        }
    }

    long getPushStartTimestamp() {
        return pushStartTimestamp;
    }

    enum PushStatus {
        OK, Fail, Timeout, Busy, ChanClosed, ChanOverflow,
    }
}
